package com.innover.activemq.config;

import com.innover.activemq.consumer.Consumer;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.listener.DefaultMessageListenerContainer;

import javax.jms.MessageListener;
import javax.jms.Queue;

/**
 * Created by James on 8/8/2017.
 */
@Configuration
public class ActiveMQConfig implements ApplicationContextAware {
    private static final Logger logger = LoggerFactory.getLogger(ActiveMQConfig.class);
    public static ApplicationContext appContext;
    private static ActiveMQSettings activeMQSettings;

    public ActiveMQConfig(@Autowired ActiveMQSettings activeMQSettings){
        this.activeMQSettings = activeMQSettings;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        appContext = applicationContext;
    }

    public static Object getBean(String beanName){
        return appContext.getBean(beanName);
    }

    public static Object getBean(Class<?> clazz){
        return appContext.getBean(clazz);
    }

    @Bean
    public ActiveMQConnectionFactory getMQFactory(){
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(activeMQSettings.getUsername(), activeMQSettings.getPassword(), activeMQSettings.getBrokerUrl());
        return connectionFactory;
    }

    @Bean(name = "standaloneTaskQueue")
    public Queue getStandaloneTaskQueue(){
        Queue queue = new ActiveMQQueue(activeMQSettings.getStandaloneTaskQueue() + "?consumer.prefetchSize=" + activeMQSettings.getPrefetch());
        return queue;
    }

    @Bean(name = "sequenceTaskQueue")
    public Queue getSequenceTaskQueue(){
        Queue queue = new ActiveMQQueue(activeMQSettings.getSequenceTaskQueue() + "?consumer.exclusive=true" + "&consumer.prefetchSize=" + activeMQSettings.getPrefetch());
        return queue;
    }

    @Bean(name = "standaloneMQConsumer")
    public MessageListener getStandaloneMQConsumer(){
        Consumer mqConsumer = new Consumer();
        return mqConsumer;
    }

    @Bean(name = "sequenceMQConsumer")
    public MessageListener getSequenceMQConsumer(){
        Consumer mqConsumer = new Consumer();
        return mqConsumer;
    }

    @Bean(name = "commonTaskQueueDefaultMessageListenerContainer")
    public DefaultMessageListenerContainer commonTaskQueueDefaultMessageListenerContainer(@Autowired ActiveMQConnectionFactory connectionFactory,
                                                                                          @Autowired @Qualifier("standaloneTaskQueue") Queue queue){
        DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
        defaultMessageListenerContainer.setConnectionFactory(connectionFactory);
        defaultMessageListenerContainer.setDestination(queue);
        defaultMessageListenerContainer.setConcurrentConsumers(activeMQSettings.getConcurrent());
        defaultMessageListenerContainer.setAutoStartup(false);
        return defaultMessageListenerContainer;
    }

    @Bean(name = "sequenceTaskQueueDefaultMessageListenerContainer")
    public DefaultMessageListenerContainer sequenceTaskQueueDefaultMessageListenerContainer(@Autowired ActiveMQConnectionFactory connectionFactory,
                                                                                            @Autowired @Qualifier("sequenceTaskQueue") Queue queue){
        DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
        defaultMessageListenerContainer.setConnectionFactory(connectionFactory);
        defaultMessageListenerContainer.setDestination(queue);
        defaultMessageListenerContainer.setConcurrentConsumers(activeMQSettings.getConcurrent());
        defaultMessageListenerContainer.setAutoStartup(false);
        return defaultMessageListenerContainer;
    }

    @Bean(name = "commonTaskQueueSelectorDefaultMessageListenerContainer")
    public DefaultMessageListenerContainer commonTaskQueueSelectorDefaultMessageListenerContainer(@Autowired ActiveMQConnectionFactory connectionFactory,
                                                                                                  @Autowired @Qualifier("standaloneTaskQueue") Queue queue){
        DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
        defaultMessageListenerContainer.setConnectionFactory(connectionFactory);
        defaultMessageListenerContainer.setDestination(queue);
        defaultMessageListenerContainer.setMessageSelector(activeMQSettings.getMessageSelector());
        defaultMessageListenerContainer.setConcurrentConsumers(activeMQSettings.getConcurrent());
        defaultMessageListenerContainer.setAutoStartup(false);
        return defaultMessageListenerContainer;
    }

    public static void startMQConsumers(){
        try{

            if(activeMQSettings.isEnableCommonConsumer()){
                DefaultMessageListenerContainer commonMLC = (DefaultMessageListenerContainer) getBean("commonTaskQueueDefaultMessageListenerContainer");
                MessageListener standaloneMQConsumer = (MessageListener) getBean("standaloneMQConsumer");
                commonMLC.setMessageListener(standaloneMQConsumer);
                commonMLC.setCacheLevel(4);
                commonMLC.afterPropertiesSet();
                commonMLC.start();
                logger.info("Start commonMLC");
            }

            if(activeMQSettings.isEnableSequenceQueueConsumer()){
                DefaultMessageListenerContainer sequenceMLC = (DefaultMessageListenerContainer) getBean("sequenceTaskQueueDefaultMessageListenerContainer");
                MessageListener sequenceMQConsumer = (MessageListener) getBean("sequenceMQConsumer");
                sequenceMLC.setMessageListener(sequenceMQConsumer);
                sequenceMLC.setCacheLevel(4);
                sequenceMLC.afterPropertiesSet();
                sequenceMLC.start();
                logger.info("Start sequenceMLC");
            }

            if(activeMQSettings.isEnableMessageSelectorConsumer()){
                DefaultMessageListenerContainer msgSelectorMLC = (DefaultMessageListenerContainer) getBean("commonTaskQueueSelectorDefaultMessageListenerContainer");
                MessageListener sequenceMQConsumer = (MessageListener) getBean("standaloneMQConsumer");
                msgSelectorMLC.setMessageListener(sequenceMQConsumer);
                msgSelectorMLC.setCacheLevel(4);
                msgSelectorMLC.afterPropertiesSet();
                msgSelectorMLC.start();
                logger.info("Start msgSelectorMLC");
            }

            logger.info("Start MQ Consumers");
        } catch (Exception e){
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }

    public static void stopMQConsumers(){
        try{

            if(activeMQSettings.isEnableCommonConsumer()){
                DefaultMessageListenerContainer commonMLC = (DefaultMessageListenerContainer) getBean("commonTaskQueueDefaultMessageListenerContainer");
                commonMLC.destroy();
            }

            if(activeMQSettings.isEnableSequenceQueueConsumer()){
                DefaultMessageListenerContainer sequenceMLC = (DefaultMessageListenerContainer) getBean("sequenceTaskQueueDefaultMessageListenerContainer");
                sequenceMLC.destroy();
            }

            if(activeMQSettings.isEnableMessageSelectorConsumer()){
                DefaultMessageListenerContainer msgSelectorMLC = (DefaultMessageListenerContainer) getBean("commonTaskQueueSelectorDefaultMessageListenerContainer");
                msgSelectorMLC.destroy();
            }

            logger.info("Stop MQ Consumers");
        } catch (Exception e){
            e.printStackTrace();
            logger.error(e.getMessage());
        }
    }
}
